package com.example.pulsar.service;

import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.client.api.*;
import org.apache.pulsar.client.api.url.URL;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
import org.apache.pulsar.common.policies.data.TenantInfo;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;

/**
 * @Auther: 141040
 * @Date: 3/25/2019 09:38
 * @Description: Pulsar生产者
 */
@Service
public class ProducerService {

    /**日志打印**/
    private final Logger logger = LoggerFactory.getLogger(this.getClass());

    @Value("${pulsarUrl}")
    private String pulsarUrl;

    @Value("${pulsarTopic:order-topic}")
    private String pulsarTopic;

    @Value("${adminApiUrl}")
    private String adminApiUrl;

    /**
     * 创建生产者
     */
    public MessageId createProducer() throws PulsarClientException {




        //1. 实例化PulsarClient客户端对象
        PulsarClient client = PulsarClient.builder().serviceUrl(pulsarUrl).build();

        //2. 判断客户端对象是否为空
        if (ObjectUtils.isEmpty(client)) {
            logger.error("创建生产者连接Pulsar客户端对象为空");
            return null;
        }

        //3. 发送消息
        String npTopic = "persistent://public/default/zhang-topic";
        Producer<byte[]> producer = client.newProducer()
                .topic(npTopic)
                .producerName("我是生产者")
                .create();
        MessageId send = producer.send("发送的Pulsar消息【pulsar测试消息内容】".getBytes());

        return send;
    }


}
